package loadData

import (
	"encoding/json"
	"fmt"
	metaD "db2s/MetaInit"
	"db2s/global"
	"db2s/limit"
	"db2s/tableEr"
	"os"
	"reflect"
	"strconv"
	"strings"
)

func indexInit(IndexData map[string]map[string]any, plan2 *SchedulePlan2, v string) {
	var (
		v1 map[string]any
		ok bool
	)
	plan2.ReadOptimizer.IndexColumnType = "missIndex"
	if v1, ok = IndexData[v]; !ok {
		return
	}
	if v2, ok1 := v1["indexColumn"]; ok1 {
		if v2 != nil {
			plan2.ReadOptimizer.IndexColumn = v2.([]string)
		}
	}
	if v2, ok1 := v1["indexType"]; ok1 {
		if v2 != nil {
			plan2.ReadOptimizer.IndexColumnType = v2.(string)
		}
	} else {
		if v3, ok2 := IndexData[v]["forceMiss"]; ok2 {
			if v3 != nil {
				plan2.ReadOptimizer.IndexColumnType = v2.(string)
			}
		}
	}
	if v2, ok1 := v1["indexColumnSeq"]; ok1 {
		if v2 != nil {
			plan2.ReadOptimizer.IndexColumnSeq = v2.(map[string]string)
		}
	}
	if v2, ok1 := v1["indexName"]; ok1 {
		if v2 != nil {
			plan2.ReadOptimizer.IndexName = v2.(string)
		}
	}
}

// table sum pod Init
func (t CheckSchemaMetaData) podInit(n any) *global.TPod {
	event := "[podInit]"
	switch GlobalPConfigs.rules.CheckMode {
	case "load":
		jsonData, err := json.Marshal(n)
		if err != nil {
			WLog.Error(fmt.Sprintf("(%d) %s map strconv json fail. error info is %v", 1, event, err))
		}
		var selectRows, insertRows, syncStatus, syncSum int64 = 0, 0, 0, 0
		var tload = global.TLoad{CheckMode: "load", SelectRows: &selectRows, InsertRows: &insertRows, SyncStatus: &syncStatus, SyncInfo: "ok", SyncSum: &syncSum}
		if err = json.Unmarshal(jsonData, &tload); err != nil {
			WLog.Error(fmt.Sprintf("(%d) %s json strconv struct fail. error info is %v", 1, event, err))
		}
		return &global.TPod{Load: tload}
	}
	return &global.TPod{}
}
func getSchemaToTableMap(s, t any) string {
	return fmt.Sprintf("%v%s%v", s, global.SchemaTableSplit, t)
}
func tatilMessageInit(plan2 *SchedulePlan2) {
	var (
		product  int64 = 0
		customer int64 = 0
		mqstart  int64 = 0
		mqlevels int64 = 0
		mqlevelr int64 = 0
	)
	plan2.mqProductCustomerMonitor = global.MqProductCustomerMonitorMessage{
		MqProductSeq:           &product,
		MqCustomerSeq:          &customer,
		MqStartTiming:          &mqstart,
		MqLevelProductSend:     &mqlevels,
		MqLevelCustomerReceive: &mqlevelr,
		ProductMessage:         make(chan map[string][]string),
	}
}

func truncate(v string, plan2 *SchedulePlan2) bool {
	fmt.Println(fmt.Sprintf("-- db2s begin truncate table %s.%s --", plan2.Object.Schema, plan2.Object.Table))
	var partitionGather []string
	for _, v1 := range GlobalPConfigs.schema.Partition[v].PartitionGather {
		if v2, ok := plan2.Object.PartitionData.SourceDestNameMap[strings.ToUpper(v1)]; ok {
			partitionGather = append(partitionGather, v2)
		}
	}
	if len(partitionGather) == 0 {
		partitionGather = append(partitionGather, "single")
	}
	return Er.TActive(Er.TableActive{
		DBType: GlobalPConfigs.dSns.SrcDBName,
	}).Truncate(global.TableActiveInput{
		LogSeq: 1,
		DB:     GlobalPConfigs.SDB,
		TableInfo: global.TableInfo{
			Schema: plan2.Object.Schema,
			Table:  plan2.Object.Table,
		},
		Input: global.StartPartConfigInputP{
			Partition: global.PartitionOptions{PartitionGather: partitionGather},
		},
		Optimizer: global.WriteOptimizer{
			SqlLogBin: GlobalPConfigs.rules.Sync.SqlLogBin,
		},
	})
}

// writeOptimizer 目标端写入的相关优化
func writeOptimizer(v string, plan2 *SchedulePlan2) {
	var thread int64 = 1
	if a, err := Er.MetaInfo(Er.TableInfoMeta{DBType: GlobalPConfigs.dSns.SrcDBName}).AvgRowLength(
		global.TablesMetaInfoInput{
			LogSeq: 1,
			DB:     GlobalPConfigs.SDB,
			TableInfo: global.TableInfo{
				Schema: plan2.Object.Schema,
				Table:  plan2.Object.Table,
			},
		}); err != nil {
		plan2.writeOptimizer.FixSum = 10
		plan2.ReadOptimizer.ChunkSum = 10
	} else {
		var b int64
		if a.(int64) > 0 {
			b = 16 * 1024 / a.(int64)
		}
		if b > 0 {
			plan2.writeOptimizer.FixSum = b
			plan2.ReadOptimizer.ChunkSum = b
		} else {
			plan2.writeOptimizer.FixSum = 10
			plan2.ReadOptimizer.ChunkSum = 10
		}
	}
	plan2.writeOptimizer.QueueData = make(chan any, GlobalPConfigs.rules.QueueSize)
	plan2.writeOptimizer.Thread = &thread
	if GlobalPConfigs.activeSwitch.TableTruncate {
		if !truncate(v, plan2) {
			fmt.Println(fmt.Sprintf("db2s report: truncate table %s.%s fail! will exit", plan2.Object.Schema, plan2.Object.Table))
			os.Exit(1)
		}
	}
}

func speedLimitInit(plan2 *SchedulePlan2) {
	var (
		readSum        int64  = 0
		writeSum       int64  = 0
		readParallel   int64  = 0
		writeParallel  int64  = 0
		readAvgTime    uint64 = 0
		writeAvgTime   uint64 = 0
		status         int64  = 0
		readExecCount  int64  = 0
		writeExecCount int64  = 0
	)
	plan2.TableSpeedLimit = global.TableSpeedLimit{
		ReadSum:        &readSum,
		WriteSum:       &writeSum,
		ReadParallel:   &readParallel,
		WriteParallel:  &writeParallel,
		ReadAvgTime:    &readAvgTime,
		WriteAvgTime:   &writeAvgTime,
		ReadExecCount:  &readExecCount,
		WriteExecCount: &writeExecCount,
		Status:         &status,
		SpeedLimit:     limit.NewRateLimiter(1000000000, 1),
		InstanceConfig: global.DBInstanceConfig{
			LowPositionConn:   GlobalPConfigs.InstanceConfig.LowPositionConn,
			CurryPositionConn: GlobalPConfigs.InstanceConfig.CurryPositionConn,
			HighPositionConn:  GlobalPConfigs.InstanceConfig.HighPositionConn,
		},
	}
}

// readOptimizer 目标端写入的相关优化
func readOptimizer(IndexData map[string]map[string]any, v string, plan2 *SchedulePlan2) {
	var (
		thread int64 = 1
	)
	indexInit(IndexData, plan2, v)
	plan2.ReadOptimizer.ChunkStartSource = "src"
	plan2.ReadOptimizer.Thread = &thread
}

// tableObject 处理库名和表名及映射问题
func tableObjectInit(v string) (string, string, string, string) {
	f0 := getSchemaTableSeparation(v)
	schema := f0[0]
	table := f0[1]
	md := metaD.GetTableStrconvO()
	md.Schema = schema
	md.Table = table
	md.TableMap = GlobalPConfigs.schema.MapTables
	mschema, mtable := metaD.SchemaTableCorver(*md)
	return schema, table, fmt.Sprintf("%v", mschema), fmt.Sprintf("%v", mtable)
}

// tableExist 判断表对象是否都存在
func tableExist(v string, s map[string]map[string]*metaD.TableStructMeta) bool {
	var (
		logSeq = 1
		event  = "[tableExist]"
	)
	//原目标端是否存在
	schema, table, mschema, mtable := tableObjectInit(v)
	continueTrue := true
	for _, v1 := range []string{v, fmt.Sprintf("%v%s%v", mschema, global.SchemaTableSplit, mtable)} {
		if _, ok := s[v1]; ok {
			continueTrue = false
		}
	}
	if continueTrue {
		WLog.Warn(fmt.Sprintf("(%d) %v table %v.%v not is exist!,will continue.", logSeq, event, schema, table))
		return true
	}
	return continueTrue
}

func objectEsaInit(plan2 *SchedulePlan2) {
	var (
		esaColumnSeq  = make(map[int]int)
		esaColumnName = make(map[string]int)
		err           error
	)
	for _, v := range plan2.Object.Esa.ColumnSlice {
		esaColumnName[v]++
	}
	for _, v := range plan2.Object.TableColData.SColumnInfo {
		if _, ok1 := esaColumnName[strings.ToLower(v.ColumnName)]; ok1 {
			var seq int
			if seq, err = strconv.Atoi(v.ColumnSeq); err != nil {
				return
			}
			esaColumnSeq[seq-1]++
		}
	}
	plan2.Object.Esa.EsaColumnSeq = esaColumnSeq
}
func objectInit(v string, s map[string]global.TableMetaGather, p global.PartitionGarth, plan2 *SchedulePlan2) {
	var (
		logSeq = 1
		event  = "[objectInit]"
	)
	schema, table, _, _ := tableObjectInit(v)
	//sShardName := t["source"][v]
	vlog := fmt.Sprintf("(%d) %s The %s data source at the %s.%s failed to initialize table structure.", logSeq, event, GlobalPConfigs.dSns.SrcDBName, schema, table)
	WLog.Debug(vlog)
	plan2.Object.Schema = schema
	plan2.Object.Table = table
	plan2.Object.TableColData = global.TableAllColumnInfoS{SColumnInfo: s[v].SourceM}
	plan2.Object.PartitionData = p
	//plan2.Object.ShardName = map[string]string{"source": sShardName}
	plan2.Object.Esa = GlobalPConfigs.schema.EncryptionProperties[v]
	objectEsaInit(plan2)
}
func tableStatusInit(plan2 *SchedulePlan2) {
	var (
		errEnd int64 = 0
		logSeq int64
	)
	plan2.Status.LogSeq = logSeq
	plan2.Status.ErrEnd = &errEnd
}

func loadDataOutputFileInit(plan *SchedulePlan2, _ int64) bool {
	var (
		filePathType string
		fileBasePath = GlobalPConfigs.rules.Load.LoadFilePath
		filePath     = fmt.Sprintf("%v/%v", fileBasePath, plan.Object.Schema)
		err          error
	)
	if filePathType, err = global.IfFileOrDir(filePath); err != nil {
		return false
	}
	switch filePathType {
	case "dir":
		//先判断路径是否存在
		if !global.FilePathIsNotExist(filePath) {
			return false
		}
		loadDataFilePath := fmt.Sprintf("%v/%v.%v", filePath, plan.Object.Table, "loadData")
		plan.load.WriteFileName = loadDataFilePath
		metaDataFilePath := fmt.Sprintf("%v/%v.%v", filePath, plan.Object.Table, "metaData")
		plan.load.WriteTableMetaFileName = metaDataFilePath
		//创建文件
		if plan.load.WriteFileN, err = global.FileIsNotExist(loadDataFilePath); err != nil {
			return false
		}
		if plan.load.WriteTableMeta, err = global.FileIsNotExist(metaDataFilePath); err != nil {
			return false
		}
	case "file":
		if plan.load.WriteFileN, err = global.FileIsNotExist(filePath); err != nil {
			return false
		}
		if plan.load.WriteTableMeta, err = global.FileIsNotExist(filePath); err != nil {
			return false
		}
	}
	if plan.load.WriteFileN == nil || plan.load.WriteTableMeta == nil {
		return false
	}
	return true
}
func loadDataInputFileInit(plan *SchedulePlan2, _ int64) bool {
	var (
		filePathType string
		filePath     = fmt.Sprintf("%v/%v", GlobalPConfigs.rules.Load.LoadFilePath, plan.Object.Schema)
		err          error
		subDir       []string
		subFileFin   *os.File
	)

	if filePathType, err = global.IfFileOrDir(filePath); err != nil {
		return false
	}
	switch filePathType {
	case "dir":
		if subDir, err = global.SubPathDirName(filePath); err != nil {
			return false
		}
		for _, v := range subDir {
			if fmt.Sprintf("%v.loadData", plan.Object.Table) == v {
				if subFileFin, err = global.SubFileIsNotExist(fmt.Sprintf("%v/%v", filePath, v)); err != nil {
					return false
				} else {
					plan.load.ReadFileN = subFileFin
					plan.load.ReadFileName = fmt.Sprintf("%v/%v", filePath, v)
				}
			}
			if fmt.Sprintf("%v.metaData", plan.Object.Table) == v {
				if subFileFin, err = global.SubFileIsNotExist(fmt.Sprintf("%v/%v", filePath, v)); err != nil {
					return false
				} else {
					plan.load.ReadTableMeta = subFileFin
					plan.load.ReadTableFileName = fmt.Sprintf("%v/%v", filePath, v)
				}
			}
		}
	case "file":
		if plan.load.ReadFileN, err = global.SubFileIsNotExist(filePath); err != nil {
			return false
		}
		if plan.load.ReadTableMeta, err = global.SubFileIsNotExist(filePath); err != nil {
			return false
		}
		plan.load.ReadTableFileName = filePath
	}
	if plan.load.ReadFileN == nil || plan.load.ReadTableMeta == nil {
		return false
	}
	return true
}

func loadDataFileInit(plan *SchedulePlan2, logSeq int64) (fileExistStatus bool) {
	if funcName, ok := loadModeFileFunc()[GlobalPConfigs.rules.Load.LoadMode]; ok {
		if funcName.Kind() == reflect.Func {
			arguments := []reflect.Value{reflect.ValueOf(plan),
				reflect.ValueOf(logSeq),
			}
			returnValue := funcName.Call(arguments)
			for i := 0; i < len(returnValue); i++ {
				fileExistStatus = returnValue[i].Interface().(bool)
			}
		}
	}
	return
}

func initLoadSchedulePlan(s *SchemaInitContext) map[string]*SchedulePlan2 {
	var (
		m = make(map[string]*SchedulePlan2)
	)
	for _, v := range s.TableList {
		var jhsum int64 = 0
		plan2 := &SchedulePlan2{
			load: global.LoadPlan{CheckMod: "count", CheckRows: global.SyncCheckCount{SrcCheckCount: &jhsum}},
		}
		tableStatusInit(plan2)
		objectInit(v, s.TcolumnMeta, s.TDistributed, s.TablePartiton[v], plan2)
		if !loadDataFileInit(plan2, 1) {
			continue
		}
		writeOptimizer(v, plan2)
		readOptimizer(s.IndexData, v, plan2)
		speedLimitInit(plan2)
		subTaskInit(plan2)
		tatilMessageInit(plan2)
		m[plan2.SchemaToTable()] = plan2
	}
	for _, v := range s.TableMiss {
		f0 := strings.Split(v, global.SchemaTableSplit)
		schema := f0[0]
		table := f0[1]
		m[fmt.Sprintf("%v.%v", schema, table)] = nil
	}
	return m
}

// InitToSchedulePlan2Table 初始化 执行计划 以 map 的形式 key 表名 value SchedulePlan2
func InitToSchedulePlan2Table() (s []map[string]*SchedulePlan2, p []string) {
	var (
		Event1       = "[InitToSchedulePlan2Table]"
		vlog         string
		logThreadSeq int64
		f2           = global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	)
	for _, k := range f2.ChanSchemaContext {
		if funcName, ok := ScheduleTaskInitFuncRef[GlobalPConfigs.rules.CheckMode]; ok {
			if funcName.Kind() == reflect.Func {
				arguments := []reflect.Value{
					reflect.ValueOf(k)}
				m1 := funcName.Call(arguments)
				var m2 = make(map[string]*SchedulePlan2)
				for i := 0; i < len(m1); i++ {
					m2 = m1[i].Interface().(map[string]*SchedulePlan2)
				}
				if len(m2) == 0 {
					vlog = fmt.Sprintf("(%d) %s Failed to initialize the table structure of the verification table at the source. There are currently no tables to validate!!!", logThreadSeq, Event1)
					WLog.Warn(vlog)
					fmt.Println("db2s report: check table is Empty!, quit quit quit please check the log for details!")
					continue
				} else {
					var m3 = make(map[string]*SchedulePlan2)
					for k1, v1 := range m2 {
						if v1 == nil {
							p = append(p, k1)
						} else {
							m3[k1] = v1
						}
					}
					s = append(s, m3)
				}
			}
		} else {

		}
	}
	if len(s) == 0 {
		return nil, nil
	}
	return
}
